package cn.doitedu.api;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/22
 * @Desc: 学大数据，上多易教育
 *   演示 mysql-cdc-connector的 source用法
 *   cdc source，能持续抓取mysql服务器上的binlog，并解析成json形式返回；
 *   返回的json，有 crud 四种类型 （见项目中的 mysql-binlog-json 文件夹）
 **/
public class _08_MysqlCdcConnector_Demo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 构造一个mysql-cdc-connector的source对象
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit46") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit46.student") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        DataStreamSource<String> stream = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc");
        stream.print();

        env.execute();


    }

}
